package com.example.aspect;

import com.example.annotation.AsyncMessageHandler;
import com.example.context.MessageHandlerContext;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.QueueInformation;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 异步消息处理切面
 */
// TODO: 2025/4/21 注意消息队外部需要下载RabbitMQ下载、安装及使用教程
@Aspect
@Component
@Slf4j
@ConditionalOnProperty(name = "spring.rabbitmq.enabled", havingValue = "true", matchIfMissing = false)
public class AsyncMessageAspect {
    
    @Autowired(required = false)
    private RabbitTemplate rabbitTemplate;
    
    @Autowired(required = false)
    private RabbitAdmin rabbitAdmin;
    
    @Autowired
    private ThreadPoolTaskExecutor messageTaskExecutor;
    
    @Autowired
    private CacheManager cacheManager;
    
    private final ConcurrentHashMap<String, MessageHandlerContext> messageContexts = new ConcurrentHashMap<>();
    
    // 消息计数器
    private final AtomicInteger messageCount = new AtomicInteger(0);
    private final AtomicInteger processingCount = new AtomicInteger(0);
    
    @Around("@annotation(asyncMessageHandler)")
    public Object handleAsyncMessage(ProceedingJoinPoint joinPoint, AsyncMessageHandler asyncMessageHandler) throws Throwable {
        // 获取方法参数
        Object[] args = joinPoint.getArgs();
        if (args.length > 0 && rabbitTemplate != null) {
            // 检查缓存
            String cacheKey = generateCacheKey(joinPoint, args[0]);
            Cache cache = cacheManager.getCache("messageCache");
            if (cache != null) {
                Cache.ValueWrapper cachedValue = cache.get(cacheKey);
                if (cachedValue != null) {
                    log.info("从缓存中获取消息结果: {}", cacheKey);
                    return cachedValue.get();
                }
            }
            
            // 创建消息属性
            MessageProperties properties = new MessageProperties();
            properties.setPriority(asyncMessageHandler.priority());
            if (asyncMessageHandler.ttl() > 0) {
                properties.setExpiration(String.valueOf(asyncMessageHandler.ttl()));
            }
            
            // 发送消息到队列
            String exchange = asyncMessageHandler.exchange();
            String routingKey = asyncMessageHandler.routingKey();
            
            if (exchange.isEmpty() && routingKey.isEmpty()) {
                rabbitTemplate.convertAndSend(asyncMessageHandler.queue(), args[0]);
            } else {
                rabbitTemplate.convertAndSend(
                    exchange.isEmpty() ? "" : exchange,
                    routingKey.isEmpty() ? asyncMessageHandler.queue() : routingKey,
                    args[0]
                );
            }
            
            // 更新消息计数
            int count = messageCount.incrementAndGet();
            if (count % 100 == 0) {
                printQueueStatus(asyncMessageHandler.queue());
            }
        }
        
        return joinPoint.proceed();
    }
    
    @RabbitListener(queues = "#{@asyncMessageHandler.queue()}")
    public void processMessage(Message message, Channel channel) {
        if (rabbitTemplate != null) {
            try {
                int processing = processingCount.incrementAndGet();
                
                // 使用线程池处理消息
                messageTaskExecutor.execute(() -> {
                    try {
                        String messageId = message.getMessageProperties().getMessageId();
                        MessageHandlerContext context = messageContexts.get(messageId);
                        if (context != null) {
                            processMessageWithRetry(context, message.getBody(), message, channel);
                        }
                    } catch (Exception e) {
                        log.error("处理消息失败", e);
                    } finally {
                        processingCount.decrementAndGet();
                    }
                });
                
                if (processing % 50 == 0) {
                    printQueueStatus(message.getMessageProperties().getConsumerQueue());
                }
            } catch (Exception e) {
                log.error("处理消息失败", e);
            }
        }
    }
    
    private void processMessageWithRetry(MessageHandlerContext context, Object request, Message message, Channel channel) {
        try {
            // 执行原始方法
            Object result = context.getJoinPoint().proceed(new Object[]{request});
            
            // 缓存结果
            String cacheKey = generateCacheKey(context.getJoinPoint(), request);
            Cache cache = cacheManager.getCache("messageCache");
            if (cache != null) {
                cache.put(cacheKey, result);
            }
            
            // 处理成功，确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            
            if (context.getAnnotation().enableLog()) {
                log.info("消息处理成功: {}", result);
            }
            
        } catch (Throwable e) {
            if (context.canRetry()) {
                context.incrementRetryCount();
                log.warn("消息处理失败，准备重试: {}", context.getRetryCount());
                
                try {
                    Thread.sleep(context.getAnnotation().retryInterval());
                    processMessageWithRetry(context, request, message, channel);
                } catch (InterruptedException ex) {
                    log.error("重试被中断", ex);
                    Thread.currentThread().interrupt();
                }
            } else {
                log.error("消息处理失败，已达到最大重试次数", e);
                try {
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                } catch (Exception ex) {
                    log.error("消息拒绝失败", ex);
                }
            }
        }
    }
    
    private String generateCacheKey(ProceedingJoinPoint joinPoint, Object request) {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        return String.format("%s.%s-%s", 
            signature.getDeclaringTypeName(),
            signature.getMethod().getName(),
            request.hashCode()
        );
    }
    
    private void printQueueStatus(String queueName) {
        if (rabbitAdmin != null) {
            try {
                QueueInformation queueInfo = rabbitAdmin.getQueueInfo(queueName);
                if (queueInfo != null) {
                    log.info("消息队列状态 - 队列: {}, 总消息数: {}, 处理中消息数: {}, 等待处理消息数: {}, 消费者数: {}",
                        queueName,
                        messageCount.get(),
                        processingCount.get(),
                        queueInfo.getMessageCount(),
                        queueInfo.getConsumerCount()
                    );
                }
            } catch (Exception e) {
                log.error("获取队列状态失败", e);
            }
        }
    }
} 